'use strict';

const { Client } = require('@elastic/elasticsearch')
var AsyncLock = require('async-lock');
const { formatDate } = require('../utils/str-helper');
var lock = new AsyncLock();

class ElasticSearchPipeline {

    open_spider(spider) {
        const server = spider.options.ds;
        const { template = {} } = spider.options.es;

        if (!server) {
            spider.log.error('You must set ES_SERVER first!')
            process.exit(1)
        }
        this.client = new Client({ node: server })
        this.template = template
    }

    async process_item(item, spider, request) {
        const { log } = spider;
        const { index_name, id, file = {}, ...body } = item;
        const { target = index_name } = spider.options;
        const rowId = id ? id : file.basename || request.reqKey;

        const _init = async () => {
            const { body: exists } = await this.client.indices.exists({ index: target })
            if (!exists) {
                await this.client.indices.create({
                    index: target,
                    body: {
                        ...this.template
                    }
                })
            }
        }

        const _addItem = async () => {
            if (target && rowId) {
                if (body.timestamp === undefined) {
                    let date = formatDate()
                    body.timestamp = date.valueOf();
                    body.date = date.format();
                }
                await this.client.index({
                    index: target,
                    id: spider.name + ":" + rowId,
                    body: {
                        _spider_name: spider.name,
                        _spider_time: new Date(),
                        _spider_uri: request.link,
                        ...file,
                        ...body
                    },
                })
                log.debug(`${target} index ${rowId} ok`)
            }
        }

        return new Promise(resolve => {
            if (target) {
                lock.acquire(
                    target,
                    function (done) {
                        _init().then(() => {
                            done();
                        })
                    },
                    function (err, ret) {
                        _addItem();
                        resolve(item);
                    },
                    {});
            } else {
                resolve(item);
            }
        })
    }

    close_spider(spider) {
        this.client && this.client.close()
    }
}

module.exports = ElasticSearchPipeline;